package pipeline

import (
	"context"
	"fmt"
	"io"
	"sync"

	"github.com/example-user/go-pocket-etl/pkg/core"
	"github.com/sirupsen/logrus"
)

// ProcessorConfig 定义了单个处理器的配置，供引擎的 Run 方法消费。
type ProcessorConfig struct {
	Type   string                 `yaml:"type"`
	Params map[string]interface{} `yaml:"params"`
}

// PipelineConfig 包含了对管道性能进行微调的参数。
// BatchSize 控制了 Sink 批量写入的大小，增大此值可提高写入吞吐量，但会增加延迟和内存消耗。
// ChannelSize 定义了连接各阶段的通道缓冲区大小，更大的缓冲区可以减少阶段间的等待，但同样会增加内存占用。
type PipelineConfig struct {
	BatchSize   int `yaml:"batch_size"`
	ChannelSize int `yaml:"channel_size"`
}

const (
	defaultBatchSize   = 100
	defaultChannelSize = 1000
)

// Engine 是 ETL 管道的并发编排器。
// 它将 Source、Processors 和 Sink 组装成一个基于 Goroutine 和 Channel 的流水线，
// 并负责管理整个流水线的生命周期，包括启动、优雅关闭和错误传播。
type Engine struct {
	source      core.Source
	processors  []core.Processor
	sink        core.Sink
	batchSize   int
	channelSize int
	cancel      context.CancelFunc // 用于在任何环节出错时，快速终止整个管道
	wg          sync.WaitGroup
}

// NewEngine 创建一个新的管道引擎实例。
func NewEngine(source core.Source, processors []core.Processor, sink core.Sink, config PipelineConfig) *Engine {
	batchSize := config.BatchSize
	if batchSize <= 0 {
		batchSize = defaultBatchSize
	}

	channelSize := config.ChannelSize
	if channelSize <= 0 {
		channelSize = defaultChannelSize
	}

	logrus.Infof("管道配置: BatchSize=%d, ChannelSize=%d", batchSize, channelSize)

	return &Engine{
		source:      source,
		processors:  processors,
		sink:        sink,
		batchSize:   batchSize,
		channelSize: channelSize,
	}
}

// Run 动态构建并启动整个并发 ETL 流水线。
func (e *Engine) Run(ctx context.Context, sourceConfig map[string]interface{}, processorConfigs []ProcessorConfig, sinkConfig map[string]interface{}) (err error) {
	// 1. 创建一个可取消的上下文，用于实现“一处失败，全体取消”的快速失败机制。
	runCtx, cancel := context.WithCancel(ctx)
	e.cancel = cancel
	defer e.cancel() // 确保在函数退出时，所有 goroutine 都能收到停止信号

	// 在函数返回前，确保所有组件的 Close() 方法都被调用。
	// 使用命名返回值和 defer 来优雅地处理组件的关闭逻辑。
	defer func() {
		logrus.Info("正在关闭数据汇 (Sink)...")
		if closeErr := e.sink.Close(); closeErr != nil && err == nil {
			err = fmt.Errorf("pipeline: failed to close sink: %w", closeErr)
		}
		for i := len(e.processors) - 1; i >= 0; i-- {
			logrus.Infof("正在关闭处理器 (Processor) #%d (%s)...", i+1, processorConfigs[i].Type)
			if closeErr := e.processors[i].Close(); closeErr != nil && err == nil {
				err = fmt.Errorf("pipeline: failed to close processor #%d (%s): %w", i+1, processorConfigs[i].Type, closeErr)
			}
		}
		logrus.Info("正在关闭数据源 (Source)...")
		if closeErr := e.source.Close(); closeErr != nil && err == nil {
			err = fmt.Errorf("pipeline: failed to close source: %w", closeErr)
		}
	}()

	// 2. 按顺序打开所有组件，这是运行前的准备和验证阶段。
	logrus.Info("正在打开数据源 (Source)...")
	if err := e.source.Open(sourceConfig); err != nil {
		return fmt.Errorf("pipeline: failed to open source: %w", err)
	}

	for i, p := range e.processors {
		logrus.Infof("正在打开处理器 (Processor) #%d (%s)...", i+1, processorConfigs[i].Type)
		if err := p.Open(processorConfigs[i].Params); err != nil {
			return fmt.Errorf("pipeline: failed to open processor #%d (%s): %w", i+1, processorConfigs[i].Type, err)
		}
	}

	logrus.Info("正在打开数据汇 (Sink)...")
	if err := e.sink.Open(sinkConfig); err != nil {
		return fmt.Errorf("pipeline: failed to open sink: %w", err)
	}

	// 3. 动态创建一系列 Channel，作为连接各个并发阶段的“传送带”。
	numWorkers := len(e.processors) + 2 // Source + Processors + Sink
	errChan := make(chan error, numWorkers)
	numChans := len(e.processors) + 1
	dataChans := make([]chan core.Record, numChans)
	for i := 0; i < numChans; i++ {
		dataChans[i] = make(chan core.Record, e.channelSize)
	}

	// 4. 为每个组件启动一个专属的 goroutine，并将它们通过 Channel 连接起来，形成流水线。
	e.wg.Add(numWorkers)
	go e.runSource(runCtx, dataChans[0], errChan)
	for i := range e.processors {
		go e.runProcessor(runCtx, e.processors[i], dataChans[i], dataChans[i+1], errChan, i+1, processorConfigs[i].Type)
	}
	go e.runSink(runCtx, dataChans[len(dataChans)-1], errChan)

	// 5. 等待所有 goroutine 执行结束。
	e.wg.Wait()
	close(errChan)

	// 6. 收集在运行过程中可能发生的任何错误，并统一返回。
	var finalErr error
	for runErr := range errChan {
		if finalErr == nil {
			finalErr = runErr
		} else {
			finalErr = fmt.Errorf("%v; %w", finalErr, runErr)
		}
	}

	return finalErr
}

// runSource 是 Source 的工作协程，负责从数据源读取数据并送入第一个通道。
func (e *Engine) runSource(ctx context.Context, outChan chan<- core.Record, errChan chan<- error) {
	defer e.wg.Done()
	defer close(outChan) // 读取完成后关闭输出通道，这是通知下游数据已耗尽的关键信号。
	logrus.Info("Source worker 启动")

	for {
		// 优先检查上下文是否已被取消，实现快速失败。
		select {
		case <-ctx.Done():
			logrus.Warn("Source worker 收到取消信号，正在停止...")
			return
		default:
			// 非阻塞地继续执行
		}

		record, err := e.source.Read()
		if err != nil {
			if err == io.EOF {
				logrus.Info("Source 已成功读取所有数据")
				return // 数据流正常结束
			}
			// 发生不可恢复的读取错误
			logrus.Errorf("Source 读取数据时发生错误: %v", err)
			errChan <- fmt.Errorf("source error: %w", err)
			e.cancel() // 发生错误，立即取消所有其他 goroutine
			return
		}

		// 将记录发送到输出通道，同时监听取消信号。
		select {
		case outChan <- record:
		case <-ctx.Done():
			logrus.Warn("Source worker 在发送数据时收到取消信号，正在停止...")
			return
		}
	}
}

// runProcessor 是流水线上的一个工作站，负责执行单个处理逻辑。
func (e *Engine) runProcessor(ctx context.Context, p core.Processor, inChan <-chan core.Record, outChan chan<- core.Record, errChan chan<- error, num int, pType string) {
	defer e.wg.Done()
	defer close(outChan) // 当前阶段处理完毕，关闭自己的输出通道，以通知下一阶段。
	logrus.Infof("Processor #%d (%s) worker 启动", num, pType)

	// for-range 会自动处理通道的关闭，是消费通道数据的优雅方式。
	for record := range inChan {
		// 每次循环开始时，都检查是否需要提前退出。
		select {
		case <-ctx.Done():
			logrus.Warnf("Processor #%d (%s) worker 收到取消信号，正在停止...", num, pType)
			return
		default:
			processedRecord, err := p.Process(record)
			if err != nil {
				logrus.Errorf("Processor #%d (%s) 处理记录时发生错误: %v", num, pType, err)
				errChan <- fmt.Errorf("processor #%d (%s) error: %w", num, pType, err)
				e.cancel()
				return
			}

			// 如果处理器返回 nil, 意味着该记录被过滤，我们通过 continue 跳过它。
			if processedRecord == nil {
				continue
			}

			select {
			case outChan <- processedRecord:
			case <-ctx.Done():
				logrus.Warnf("Processor #%d (%s) worker 在发送数据时收到取消信号，正在停止...", num, pType)
				return
			}
		}
	}
	logrus.Infof("Processor #%d (%s) worker 输入通道已关闭，正常退出", num, pType)
}

// runSink 是 Sink 的工作协程，负责从最后一个通道接收数据并批量写入目的地。
func (e *Engine) runSink(ctx context.Context, inChan <-chan core.Record, errChan chan<- error) {
	defer e.wg.Done()
	logrus.Info("Sink worker 启动")
	batch := make([]core.Record, 0, e.batchSize)

	for record := range inChan {
		// 同样，优先检查取消信号。
		select {
		case <-ctx.Done():
			logrus.Warn("Sink worker 收到取消信号，正在停止...")
			return
		default:
			batch = append(batch, record)
			if len(batch) >= e.batchSize {
				logrus.Infof("正在刷入一批 %d 条记录...", len(batch))
				if err := e.flush(batch); err != nil {
					logrus.Errorf("Sink 刷入批次时发生错误: %v", err)
					errChan <- fmt.Errorf("sink error: %w", err)
					e.cancel()
					// 注意：这里在出错后没有立即 return，是为了让循环自然结束，
					// 从而可以继续处理 defer 和最后的 flush。
					// 但因为 cancel() 被调用，其他 goroutine 会快速退出。
				}
				batch = make([]core.Record, 0, e.batchSize) // 重置批次
			}
		}
	}

	// 注意：循环结束后，必须处理最后一批可能不足一个 batchSize 的数据，否则会造成数据丢失。
	if len(batch) > 0 {
		logrus.Infof("正在刷入最后 %d 条记录...", len(batch))
		if err := e.flush(batch); err != nil {
			logrus.Errorf("Sink 刷入最后批次时发生错误: %v", err)
			errChan <- fmt.Errorf("sink error on final flush: %w", err)
		}
	}
	logrus.Info("Sink worker 输入通道已关闭，正常退出")
}

// flush 将一个批次的数据写入 sink。
func (e *Engine) flush(batch []core.Record) error {
	if len(batch) == 0 {
		return nil
	}
	return e.sink.Write(batch)
}
